package com.bs.flinktest.config;

import org.apache.flink.api.common.JobStatus;
import org.apache.flink.core.execution.JobClient;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import com.bs.flinktest.vo.TestDataSource;

import javax.annotation.PostConstruct;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author bingshao
 * @date 2023/2/1
 **/
@Configuration
public class SqlCdc {


    @PostConstruct
    public void initSqlFlink() throws Exception {
        ThreadFactory namedThreadFactory = new CustomizableThreadFactory("task-pool-");

        ExecutorService executor = new ThreadPoolExecutor(5, 10, 5L, TimeUnit.MINUTES, new LinkedBlockingQueue<>(600), namedThreadFactory, new ThreadPoolExecutor.DiscardOldestPolicy());

        List<TestDataSource> testDataSources = new LinkedList<>();
        TestDataSource testDataSource1 = new TestDataSource();
        testDataSource1.setHostName("110.42.233.197");
        testDataSource1.setPort(3306);
        testDataSource1.setUsername("root");
        testDataSource1.setPassword("bs1354361838");
        testDataSource1.setDatabaseName("bingshao");
        testDataSource1.setTableName("user");
        testDataSource1.setTargetTableName("user_copy");
        testDataSource1.setTargetUrl("jdbc:mysql://110.42.233.197:3306/bingshao?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai");
        testDataSource1.setTargetUsername("root");
        testDataSource1.setTargetPassword("bs1354361838");
        testDataSource1.setDriver("com.mysql.cj.jdbc.Driver");

        TestDataSource testDataSource2 = new TestDataSource();
        testDataSource2.setHostName("110.42.233.197");
        testDataSource2.setPort(3306);
        testDataSource2.setUsername("root");
        testDataSource2.setPassword("bs1354361838");
        testDataSource2.setDatabaseName("bingshao");
        testDataSource2.setTableName("order");
        testDataSource2.setTargetTableName("order_copy");
        testDataSource2.setTargetUrl("jdbc:mysql://110.42.233.197:3306/bingshao?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai");
        testDataSource2.setTargetUsername("root");
        testDataSource2.setTargetPassword("bs1354361838");
        testDataSource2.setDriver("com.mysql.cj.jdbc.Driver");

        testDataSources.add(testDataSource1);
        testDataSources.add(testDataSource2);

        //            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //            env.enableCheckpointing(60 * 1000L);
        //            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30 * 1000L);
        //            env.getCheckpointConfig().setCheckpointTimeout(60 * 2 * 1000L);
        //            env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        ////        env.getCheckpointConfig().setCheckpointStorage("file:///checkpoint");
        //            StreamTableEnvironment ste = StreamTableEnvironment.create(env);
        //            ste.executeSql(
        //                    "create table `mysql" + testDataSource.getTableName() + "`" + "( id bigint,name VARCHAR," +
        //                            "    PRIMARY KEY (id) NOT ENFORCED" +
        //                            ") WITH (" +
        //                            " 'scan.incremental.snapshot.enabled' = 'false'," +
        ////                        " 'scan.startup.mode' = 'initial', "+
        //                            " 'connector' = 'mysql-cdc'," +
        //                            " 'hostname' = '" + testDataSource.getHostName() + "'," +
        //                            " 'port' = '" + testDataSource.getPort() + "'," +
        //                            " 'username' = '" + testDataSource.getUsername() + "'," +
        //                            " 'password' = '" + testDataSource.getPassword() + "'," +
        //                            " 'database-name' = '" + testDataSource.getDatabaseName() + "'," +
        //                            " 'table-name' = '" + testDataSource.getTableName() + "'" +
        //                            ")"
        //            );
        ////            ste.executeSql(
        ////                    "create table mysqlorder( id bigint,user_id bigint,product_name VARCHAR," +
        ////                            "    PRIMARY KEY (id) NOT ENFORCED" +
        ////                            ") WITH (" +
        ////                            " 'scan.incremental.snapshot.enabled' = 'false'," +
        ////                            " 'connector' = 'mysql-cdc'," +
        ////                            " 'hostname' = '110.42.233.197'," +
        ////                            " 'port' = '3306'," +
        ////                            " 'username' = 'root'," +
        ////                            " 'password' = 'bs1354361838'," +
        ////                            " 'database-name' = 'bingshao'," +
        ////                            " 'table-name' = 'order'" +
        ////                            ")"
        ////            );
        //
        //            ste.executeSql(
        //                    "create table `mysqlsink" + testDataSource.getTableName() + "`" + "(" +
        //                            "id bigint,name VARCHAR," +
        //                            "    PRIMARY KEY (id) NOT ENFORCED" +
        //                            ") WITH (" +
        //                            "'connector' = 'jdbc'," +
        //                            "'url' = '" + testDataSource.getTargetUrl() + "'," +
        //                            "'username' = '" + testDataSource.getTargetUsername() + "'," +
        //                            "'password' = '" + testDataSource.getTargetPassword() + "'," +
        //                            "'table-name' = '" + testDataSource.getTargetTableName() + "'," +
        //                            "'driver' = '" + testDataSource.getDriver() + "'" +
        //                            ")"
        //            );
        //
        //
        //            ste.executeSql(
        //                    "insert into `mysqlsink" + testDataSource.getTableName() + "`" + "(id,name) select id,name from `mysql" + testDataSource.getTableName() + "`"
        //            );
        testDataSources.forEach(testDataSource -> {
            Future<JobClient> future = executor.submit(testDataSource);
            try {
                JobStatus jobStatus = future.get().getJobStatus().get();
                System.out.println("jobStatus===========" + jobStatus);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        });


    }


}
